package com.alibaba.bink;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2817:35
 */
public class 过滤算子 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines = env.socketTextStream("galeno", 11111);
        SingleOutputStreamOperator<String> filter = lines.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                if (value.contains("a")) {
                    return true;
                }
                return false;
            }
        });
        filter.print();
        System.out.println(env.getExecutionPlan());
        env.execute();


    }
}
